package com.stox.services.core.repository;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.StringWriter;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.stox.shared.core.util.Constant;
import com.stox.shared.core.util.PathUtil;

public abstract class AbstractPartitioningFileRepository<T, V> implements Runnable {
	
	public static interface EntityIterator<T> {
		public boolean shouldContinue(T entity);
	}

	private boolean dirty;
	
	private long flushableEntityCount = 500000;
	
	private Map<String, StringWriter> writers = new HashMap<String, StringWriter>();

	protected abstract T deserialize(String text, V partitionModel);

	protected abstract String serialize(T entity);

	protected abstract String getFilePath(V partitionModel);

	public void setFlushableEntityCount(long flushableEntityCount){
		this.flushableEntityCount = flushableEntityCount;
	}
	
	public String getFilebasePath() {
		return PathUtil.getRootPath() + File.separator + "filebase";
	}

	@Override
	public void run() {
		try {
			flush();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public void flush() {
		if(!dirty) return;
		System.out.println("Flushing...");
		ThreadPoolExecutor executor = new ThreadPoolExecutor(20,20,100, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
		for(final  String filePath : writers.keySet()){
			executor.execute(new Runnable(){
				@Override
				public void run() {
					try {
						FileWriter fileWriter = new FileWriter(filePath,true);
						StringWriter stringWriter = writers.get(filePath);
						fileWriter.append(stringWriter.toString());
						fileWriter.flush();
						fileWriter.close();
						stringWriter.getBuffer().setLength(0);
					} catch (IOException e) {
						e.printStackTrace();
					}
				}
			});
		}
		
		try {
			executor.shutdown();
			executor.awaitTermination(1, TimeUnit.MINUTES);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}finally{
			dirty = false;
		}
		
	}

	protected void iterate(EntityIterator<T> iterator, V partitionModel) {
		BufferedReader bufferedReader = null;
		try {
			bufferedReader = new BufferedReader(new InputStreamReader(
					new ReverseLineInputStream(new File(
							getFilePath(partitionModel)))));
			String line = null;
			while ((line = bufferedReader.readLine()) != null
					&& (null == iterator || iterator
							.shouldContinue(deserialize(line, partitionModel)))) {
			}
		} catch (Exception e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} finally {
			if (null != bufferedReader) {
				try {
					bufferedReader.close();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
	}

	private long entityCounter;
	public T save(T entity, V partitionModel) {
		StringWriter writer = writers.get(getFilePath(partitionModel));
		if (null == writer) {
			createPartition(partitionModel);
			writer = writers.get(getFilePath(partitionModel));
		}
		synchronized (writer) {
			writer.write(entity.toString() + Constant.lineSeparator);
		}
		if(++entityCounter >= flushableEntityCount ){
			flush();
			entityCounter = 0;
		}else{
			dirty = true;
		}
		return entity;
	}

	public void createPartition(V partitionModel) {
		try {
			String filePath = getFilePath(partitionModel);
			if (null == filePath)
				return;
			File file = new File(filePath);
			if (!file.exists()) {
				file.getParentFile().mkdirs();
				file.createNewFile();
			}
			writers.put(filePath, new StringWriter());
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

}
